package com.desheng.bigdata.flink.cep

import java.util

import org.apache.flink.cep.PatternSelectFunction
import org.apache.flink.cep.scala.CEP
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}
import org.apache.flink.streaming.api.windowing.time.Time
/**
  * flink复杂事件处理
  *     连续在10s内登陆失败，需要被提取出来，进行人工的分析，是否会有一些危害
  */
object FlinkLoginEventCepOps {
    def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        val loginEventStream:DataStream[LoginEvent] = env.fromCollection(List(
            LoginEvent("1", "192.168.0.1", "fail", "1558430842"),
            LoginEvent("1", "192.168.0.2", "fail", "1558430843"),
            LoginEvent("1", "192.168.0.3", "fail", "1558430844"),
            LoginEvent("2", "192.168.10.10", "success", "1558430845")
        ))
        val user2LoginEvent = loginEventStream.keyBy(loginEvent => loginEvent.userId)
        //定义匹配的cep规则 Pattern不是jdk中内容
        val pattern = Pattern.begin[LoginEvent]("begin")
                    .where(event => event.eventType == "fail")//第一次登陆失败
                    .next("next")
                    .where(event => event.eventType == "fail")//第二次登陆失败
                    .within(Time.seconds(10)) //在10s内连续两次登陆失败
        //应用或者检测规则
        val patternStream = CEP.pattern(user2LoginEvent, pattern)

        //获取满足规则的数据
        patternStream.select(new AlertLoginSelectFunction).print()

        env.execute()
    }
}
//封装用户登陆事件的case class
case class LoginEvent(userId: String, ip: String, eventType: String, eventTime: String)
//用于封装满足规则数据之后的警告信息
case class Warning(userId: String, firstEventTime: String, secondEventTime: String, msg: String)

class AlertLoginSelectFunction extends PatternSelectFunction[LoginEvent, Warning] {
    override def select(pattern: util.Map[String, util.List[LoginEvent]]): Warning = {
        val beginFail = pattern.getOrDefault("begin", null).iterator().next()
        val nextFail = pattern.getOrDefault("next", null).iterator().next()
        Warning(beginFail.userId, beginFail.eventTime, nextFail.eventTime, "连续两次登陆失败，关进小黑屋5分钟~")
    }
}